#include "../common/net.hpp"
#include <unordered_set>
#include <unordered_map>
namespace myrpc
{
    namespace server
    {
        class TopicManager
        {
            public:
                using ptr=std::shared_ptr<TopicManager>;
                TopicManager(){}
                void onTopicRequest(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                {
                    bool ret=true;
                    TopicOptype topic_optype=msg->topicOptype();
                    switch(topic_optype)
                    {
                        //主题创建
                        case TopicOptype::TOPIC_CREATE: topicCreate(conn,msg);break;
                        //取消订阅主题
                        case TopicOptype::TOPIC_CANCEL: topicCancel(conn,msg);break;
                        //主题删除
                        case TopicOptype::TOPIC_REMOVE: topicRemove(conn,msg);break;
                        //主题消息的发布
                        case TopicOptype::TOPIC_PUBLISH: ret=topicPublish(conn,msg);break;
                        //主题订阅
                        case TopicOptype::TOPIC_SUBSCRIBE: ret=topicSubscribe(conn,msg);break;
                        default: return errorResponse(conn,msg,RCode::RCODE_INVALID_OPTYPE);
                    }
                    if(!ret)
                        return errorResponse(conn,msg,RCode::RCODE_INVALID_OPTYPE);
                    return topicResponse(conn,msg);
                }
                //一个订阅者在连接断开时的处理--删除其管理的数据
                void onShutDown(const BaseConnection::ptr &conn)
                {
                    //消息发布者断开连接，不需要任何操作，消息订阅者断开连接需要删除其管理信息
                    //1.判断断开连接的是否是订阅者，不是的话直接返回
                    std::vector<Topic::ptr> topics;
                    Subscriber::ptr subscriber;
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        auto it=_subscribers.find(conn);
                        if(it==_subscribers.end())
                        {
                            return;
                        }
                        //2.获取到这个断开连接的订阅者，所订阅的主题对象
                        subscriber=it->second;
                        for(auto &topic_name:subscriber->_topics)
                        {
                            auto topic_it=_topics.find(topic_name);
                            if(topic_it==_topics.end())
                            {
                                continue;
                            }
                            topics.push_back(topic_it->second);
                        }
                        //3.从订阅者管理信息中删除该订阅者
                        _subscribers.erase(it);
                    }
                    //将被订阅的主题对象中，存储的该订阅者信息删除
                    for(auto &topic:topics)
                    {
                        topic->removeSubscriber(subscriber);
                    }
                }
            private:
                void errorResponse(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg,RCode rcode)
                {
                    auto msg_rsp=MessageFactory::create<TopicResponse>();
                    msg_rsp->setId(msg->rid());
                    msg_rsp->setMType(MType::RSP_TOPIC);
                    msg_rsp->setRcode(rcode);
                    conn->send(msg_rsp);
                }
                void topicResponse(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                {
                    auto msg_rsp=MessageFactory::create<TopicResponse>();
                    msg_rsp->setId(msg->rid());
                    msg_rsp->setMType(MType::RSP_TOPIC);
                    msg_rsp->setRcode(RCode::RCODE_OK);
                    conn->send(msg_rsp);
                }
            private:
                    //主题创建
                    void topicCreate(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                    {
                        std::unique_lock<std::mutex> lock(_mutex);

                        auto topic_name=msg->topicKey();
                        auto topic=std::make_shared<Topic>(topic_name);

                        _topics.insert(std::make_pair(topic_name,topic));
                    }
                    //主题删除
                    void topicRemove(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                    {
                        //1、查看当前主题，有哪些订阅者，然后从订阅者中将主题信息删除掉
                        //2、删除主题的数据 --主题名称与主题对象的映射关系
                        auto topic_name=msg->topicKey();
                        std::unordered_set<Subscriber::ptr> subscribers;
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            //通过<主题，主题对象>_topics映射，找到删除的主题对象
                            auto it=_topics.find(topic_name);
                            if(it==_topics.end())
                            {
                                return;
                            }
                            //在找到的主题对象中，主题对象中存储的订阅者,找到对应的订阅者，将订阅者本身存储的订阅主题信息中，这个删除的主题信息删掉
                            subscribers=it->second->_subscribers;
                            _topics.erase(it);
                        }
                        for(auto &subscriber:subscribers)
                        {
                            subscriber->removeTopic(topic_name);
                        }
    
                    }
                    //主题订阅
                    bool topicSubscribe(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                    {
                        //1.先找到主题对象，以及订阅者对象
                        //如果主题对象不存在，则返回false
                        //如果订阅者对象不存在，就代表是第一次连接，新建一个订阅者对象
                        Topic::ptr topic;
                        Subscriber::ptr subscriber;
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            auto topic_it=_topics.find(msg->topicKey());
                            if(topic_it==_topics.end())
                            {
                                return false;
                            }
                            topic=topic_it->second;
                            auto subscriber_it=_subscribers.find(conn);
                            if(subscriber_it==_subscribers.end())
                            {
                                subscriber=std::make_shared<Subscriber>(conn);
                                _subscribers.insert(std::make_pair(conn,subscriber));
                            }
                            else{
                                subscriber=subscriber_it->second;
                            }
                        }
                        //2.在主题对象中，添加订阅者；在订阅者对象中，添加一个订阅的主题的名称
                        topic->appendSubscriber(subscriber);
                        subscriber->appendTopic(msg->topicKey());
                        return true;
                    }
                    //取消订阅主题
                    void topicCancel(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                    {
                        //1.先找到主题对象，以及订阅者对象
                        Topic::ptr topic;
                        Subscriber::ptr subscriber;
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            auto topic_it=_topics.find(msg->topicKey());
                            if(topic_it!=_topics.end())
                            {
                                topic=topic_it->second;
                            }
                            auto subscriber_it=_subscribers.find(conn);
                            if(subscriber_it!=_subscribers.end())
                            {
                                subscriber=subscriber_it->second;
                            }
                        }   
                        //2.从主题对象中，删除当前订阅者的连接；从订阅者对象中，删除所订阅的主题名称
                        if(topic)   
                            topic->removeSubscriber(subscriber);
                        if(subscriber)  
                            subscriber->removeTopic(msg->topicKey());                     
                    }
                    //主题消息发布
                    bool topicPublish(const BaseConnection::ptr &conn,const TopicRequest::ptr &msg)
                    {    
                       //1.先找到主题对象
                        Topic::ptr topic;
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            auto topic_it=_topics.find(msg->topicKey());
                            if(topic_it!=_topics.end())
                            {
                                topic=topic_it->second;
                            }
                            else{
                                return false;
                            }
                        }   
                        topic->pushMessage(msg);
                      
                        return true;                    
                    }                
            private:
                struct Subscriber
                {
                    public:
                        using ptr=std::shared_ptr<Subscriber>;
                        std::mutex _mutex;
                        BaseConnection::ptr _conn;
                        std::unordered_set<std::string> _topics;//订阅者所订阅的主题名称


                        Subscriber(const BaseConnection::ptr &conn)
                            :_conn(conn)
                        {}
                        //订阅主题时调用
                        void appendTopic(const std::string &topic_name)
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            _topics.insert(topic_name);
                        }
                        //主题被删除 或者取消订阅的时候调用
                        void removeTopic(const std::string &topic_name)
                        {                           
                            std::unique_lock<std::mutex> lock(_mutex);
                            _topics.erase(topic_name);
                        }
                };    
                struct Topic{
                    using ptr=std::shared_ptr<Topic>;
                    std::mutex _mutex;
                    std::string _topic_name;
                    std::unordered_set<Subscriber::ptr> _subscribers;//当前主题的订阅者

                    Topic(const std::string &topic_name)
                        :_topic_name(topic_name)
                    {}
                    //新增订阅者时调用
                    void appendSubscriber(const Subscriber::ptr &subscriber)
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        _subscribers.insert(subscriber);
                    }
                    //取消订阅 或 订阅者断开连接时调用
                    void removeSubscriber(const Subscriber::ptr &subscriber)
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        _subscribers.erase(subscriber);
                    }
                    //向_subscribers的所有订阅者发布消息
                    void pushMessage(const BaseMessage::ptr &msg) 
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        for(auto &subscriber:_subscribers)
                        {
                            subscriber->_conn->send(msg);
                        }
                    }
                };
                

            private:
                std::mutex _mutex;
                std::unordered_map<std::string,Topic::ptr> _topics;//通过主题名字找到对应主题对象
                std::unordered_map<BaseConnection::ptr,Subscriber::ptr> _subscribers;
        };
    }
}